package com.share;

import java.io.FileWriter;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @desc: 实现1000000个请求，队列每次能放10000个id，消费者消费完剩余1000则补充9000，直到1000000个用完
 * 1621755031575
 * 1621755006068
 * @author:caifan
 * @date:2021/5/22
 */
public class ArrayBlockingQueueMain {
    private static final Integer CAPACITY = 10000000;
    private final static int FILL_CAPACITY = 2000;
    private static final int DEFAULT_CAPACITY = 10000;
    private static final ArrayBlockingQueue<Integer> arrayBlockingQueue = new ArrayBlockingQueue<>(CAPACITY);

    private static Lock lock0 = new ReentrantLock();
    private static Lock lock1 = new ReentrantLock();

    private static CountDownLatch latch = new CountDownLatch(15);

    private volatile static int Max = 0;

    public static void main(String[] args) throws Exception {
        long start = System.currentTimeMillis();
//        PrintWriter pw = new PrintWriter("D:/test/aaa.txt");
//        pw.append("start:" + start + "\n");
//        ExecutorService executorService = Executors.newFixedThreadPool(20);
        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(10, 10, 0,
                TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(40));
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);

//        for (int i = 0; i < 30; i++) {
//            executorService.execute(new Producer());
//        }
        for (int i = 0; i < 5; i++) {
            scheduledExecutorService.
                    scheduleAtFixedRate(new Producer(latch), 0, 100, TimeUnit.MILLISECONDS);
        }
        for (int j = 0; j < 10; j++) {
            poolExecutor.execute(new Consumer(latch));
        }
//        pw.append("end:" + System.currentTimeMillis());
//        pw.flush();

        /*for (int i = 1; i <= CAPACITY; i++) {
            arrayBlockingQueue.add(i);
        }
        long end = System.currentTimeMillis();
        System.out.println(arrayBlockingQueue.take());*/
        latch.await();
        poolExecutor.shutdown();
        scheduledExecutorService.shutdown();
        long end = System.currentTimeMillis();
        Thread.sleep(1000);
        System.err.println("耗时：" + (end - start) + "ms");
    }


    static class Producer implements Runnable {
        private CountDownLatch latch;
        Producer(CountDownLatch latch) {
            this.latch = latch;
        }

        @Override
        public void run() {
            lock0.lock();
            try {
                while (Max < CAPACITY) {
                    if (arrayBlockingQueue.size() < FILL_CAPACITY) {
                        for (int i = Max; i < Max + FILL_CAPACITY; i++) {
                            arrayBlockingQueue.offer(i);
                        }
                        Max = Max + FILL_CAPACITY;
                    }
                }
                latch.countDown();
            } finally {
                lock0.unlock();
            }
        }
    }

    static class Consumer implements Runnable {
        private CountDownLatch latch;
        Consumer(CountDownLatch latch) {
            this.latch = latch;
        }
        @Override
        public void run() {
//            FileWriter fw = null;
            try {
                while (arrayBlockingQueue.size() > 0) {
                    lock1.tryLock(3, TimeUnit.SECONDS);

                    FileWriter fw = new FileWriter("D:/test/abc.txt", true);
                    try {
                        for (int i = 0; i < 1000; i++) {
//                            Integer poll = arrayBlockingQueue.take();
                            Integer poll = arrayBlockingQueue.poll();
                            if (Objects.nonNull(poll)) {
                                System.out.println(Thread.currentThread().getName() + "::" + poll);
                                fw.write(poll + "\n");
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        fw.flush();
                        fw.close();
                        lock1.unlock();
                    }
                }
                latch.countDown();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}

